package cn.itcast.nems.listener;

import cn.hutool.core.lang.Assert;
import cn.itcast.nems.common.account.AccountHolder;
import cn.itcast.nems.common.account.AccountInfo;
import cn.itcast.nems.common.tenant.TenantInfo;
import cn.itcast.nems.common.tenant.TenantInfoHolder;
import cn.itcast.nems.manager.msas.MsasManager;
import cn.itcast.nems.manager.redis.RedisUtil;
import cn.itcast.nems.manager.redis.constant.RedisKeyConstant;
import cn.itcast.nems.manager.util.CustomBeanUtil;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.itcast.msas.sdk.acm.domain.AccountDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.nio.charset.StandardCharsets;

/**
 * <p>
 * 消息监听
 * </p>
 *
 * @author zhouwei
 * @since 2023/10/26 11:40 AM
 **/
@Slf4j
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}",
        consumerGroup = "${rocketmq.consumer.group}",
        consumeMode = ConsumeMode.ORDERLY,
        maxReconsumeTimes = 15,
        consumeTimeout = 180)
@Component
public class OrderMessageListener implements RocketMQListener<MessageExt> {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private MessageListenerProcessor messageListenerProcessor;
    @Autowired
    private MsasManager msasManager;
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void onMessage(MessageExt messageExt) {
        Assert.notNull(messageExt, "消息体为空! ");
        log.info("【消息监听】进入消费逻辑 ，TAGS: {}，消息体详情: {}", messageExt.getTags(), messageExt);
        //  通过 检查 messageId 有没有消费过，  避免多次消费产生重复数据， 实现数据库操作幂等性
        String redisKey = CustomBeanUtil.append("|", RedisKeyConstant.REDIS_ORDER_MESSAGE_ID, messageExt.getMsgId());
        try {
            if (redisUtil.hasKey(redisKey)) {
                log.info("【消息监听】msgId：{} ，消息体：{} 已成功消费！", messageExt.getMsgId(), JSON.toJSONString(messageExt));
                return;
            }
            String payload = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            log.info("【消息监听】进入消费逻辑, msg: {}", payload);
            this.setAccountTenant(payload);
            this.processMessage(messageExt, payload);
            redisUtil.set(redisKey, 1, 60 * 60 * 3);
        } catch (Exception e) {
            log.warn(e.getMessage(), e);
            redisUtil.del(redisKey);
            throw new RuntimeException(e);
        } finally {
            AccountHolder.clear();
            TenantInfoHolder.clear();
        }
    }

    private void setAccountTenant(String payload) throws JsonProcessingException {
        final JsonNode jsonNode = objectMapper.readTree(payload);
        String email = jsonNode.path("operatorEmail").asText("");
        String name = "";
        String number = "";
        if (StringUtils.hasText(email)) {
            final AccountDto account = this.msasManager.getAccount(email);
            if (account != null) {
                name = account.getName();
                number = account.getEmployeeNumber();
            } else {
                logger.warn("email '{}' not found account", email);
                email = "";
            }
        } else {
            logger.warn("msg: {} not found operatorEmail", payload);
        }
        AccountHolder.setAccountInfo(new AccountInfo(email, name, number, ""));
        TenantInfoHolder.setTenantInfo(TenantInfo.DEFAULT_TENANT_INFO);
    }

    private void processMessage(MessageExt messageExt, String payload) {
        Assert.notBlank(messageExt.getTags(), "【消息监听】TAGS 为空! ");
        String tag = messageExt.getTags().toUpperCase();
        log.info("【消息监听】进入消费逻辑, msg: {}", payload);
        messageListenerProcessor.process(tag, payload);
    }
}
